package scala.hdfs.producer

import java.util.{Date, Properties}

import cn.getech.data.development.utils.DateUtils
import com.google.gson.Gson
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

object FlinkMutilDAGProducer {

  val actions = List[String]("download", "export", "click")
  val gson = new Gson()

  val topic = "test1213"

  def main(args: Array[String]): Unit = {
    kafkaSourceProducer()
  }

  def kafkaSourceProducer(): Unit = {
//    val servers = "bigdata-test-4:9092,bigdata-test-5:9092,bigdata-test-6:9092"

    val servers = "cdh-hadoop-2:9092"
    val acks = "-1"
    val keySer = "org.apache.kafka.common.serialization.StringSerializer"
    val valueSer = "org.apache.kafka.common.serialization.StringSerializer"
    val properties = new Properties()
    properties.put("bootstrap.servers", servers)
    properties.put("acks", acks)
    properties.put("key.serializer", keySer)
    properties.put("value.serializer", valueSer)

    val producer = new KafkaProducer[String, String](properties)
    var i = 100
    while (i < 100000) {
      i += 1
      val userString = getUserBehavior(i)
      println(userString)
      val record: ProducerRecord[String, String] = new ProducerRecord[String, String](topic, s"${(System.currentTimeMillis() % 3).toString}", userString)
      producer.send(record)
//      Thread.sleep(100L)
    }
    producer.close()
  }

  def getUserBehavior(index: Int) = {
    val timestamp = System.currentTimeMillis()
    val arr = Array("exposure", "download")
    val useridNum = index

    val user = RecActive(SubModel(MSubModel(new DateUtils().getTodayDate), "sub_action"), arr((timestamp % 2).toInt), timestamp % 9, timestamp / 1000)// DateUtils.getTodayDateTime.toLong) //, DateUtils.getTodayDate)
    val str = gson.toJson(user, classOf[RecActive])
    str

  }

  case class RecActive(userid: SubModel, action: String, duration: Long, dt: Long) //, ts: String)
  case class SubModel(userid_2: MSubModel, action: String) //, ts: String)
  case class MSubModel(userid_3: String) //, ts: String)

}
